#ifndef _REACTOR_HPP
#define _REACTOR_HPP

#include <unistd.h>
#include <sys/epoll.h>
#include <vector>
#include <map>
#include "Client.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <iostream>
#include <cstring>
#include <chrono>
#include "NetEvent.hpp"

using namespace std;

class Reactor {
public:
    explicit Reactor(NetEvent* ne, int reactorId) : _ne(ne), _reactorId(reactorId) {}

    ~Reactor() {
        close(_epollfd);
        _ne = nullptr;
        _clientBuffer.clear();
        delete[] _evs;
        for (auto &itr : _clients) {
            delete itr.second;
        }
        _clients.clear();
    }

    void Start() {
        std::lock_guard<std::mutex> lock(_mutex);
        if (!_runFlag) {
            _runFlag = true;
            std::thread t(&Reactor::Run, this);
            t.detach();
        }
    }

    void Run() {
        thread::id tid = std::this_thread::get_id();
        cout << "connection thread start cid:" << tid << endl;
        _epollfd = epoll_create(1024);
        if (_epollfd == -1) {
            cout << "Reactor Run epoll_create() error" << strerror(errno) << endl;
        }
        _evs = new epoll_event[_mEvents];
        while (_runFlag) {
            if (!_clientBuffer.empty()) {
                std::lock_guard<std::mutex> lock(_mutex);
                for (auto &c : _clientBuffer) {
                    _clients[c->Sockfd()] = c;
                    OnClientJoin(c);
                }
                _clientBuffer.clear();
            }
            if (_clients.empty()) {
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
                continue;
            }

            for (auto &c : _clients) {
                epoll_event ev;
                ev.data.ptr = c.second;
                if (c.second->NeedWrite()) {
                    ev.events = EPOLLIN|EPOLLOUT;
                } else {
                    ev.events = EPOLLIN;
                }
                epoll_ctl(_epollfd, EPOLL_CTL_MOD, c.second->Sockfd(), &ev);
            }

            int ret = epoll_wait(_epollfd, _evs, _mEvents, 1);
            if (ret < 0) {
                if (ret == -1) {
                    if (errno == EINTR) continue;
                    cout << "Reactor Run epoll_wait() error" << strerror(errno) << endl;
                     break;
                }
            } else {
                for (int i = 0; i < ret; i++) {
                    auto c = (Client*)_evs[i].data.ptr;
                    if (_evs[i].events & EPOLLIN) {
                        if (-1 == c->RecvData()) {
                            RemoveClient(c);
                        } else {
                            HandleMessage(c);
                        }
                    }
                    if (_evs[i].events & EPOLLOUT) {
                        if (-1 == c->WriteData()) {
                            RemoveClient(c);
                        }
                    }
                }
            }
        }
        std::lock_guard<std::mutex> lock(_mutex);
        cout << "connection thread quit" << endl;
        _exitStatus = 1;
        _cond.notify_one();
    }

    void AddClient(Client* c) {
        std::lock_guard<std::mutex> lock(_mutex);
        _clientBuffer.push_back(c);
    }

    void Exit() {
        std::unique_lock<std::mutex> lock(_mutex);
        if (_runFlag) {
            _runFlag = false;
            _cond.wait(lock, [this] {
                return _exitStatus == 1;
            });
            delete this;
        }
    }

    bool findClient(int sockfd) {
        return _clients.count(sockfd);
    }

    void RemoveClient(Client* c) {
        cout << "thread_id:" << this_thread::get_id() << endl;
        if (_clients.count(c->Sockfd())) {
            epoll_ctl(_epollfd, EPOLL_CTL_DEL, c->Sockfd(), nullptr);
            close(c->Sockfd());
            _ne->onClose(c);
            _clients.erase(c->Sockfd());
            cout << c->Sockfd() << endl;
            delete c;
        }
    }

private:
    void OnClientJoin(Client* c) {
        epoll_event ev;
        ev.events = EPOLLIN;
        ev.data.ptr = c;
        epoll_ctl(_epollfd, EPOLL_CTL_ADD, c->Sockfd(), &ev);
        _ne->onConnect(c);
    }

    void HandleMessage(Client* c) {
        while (c->DataLen()) {
            _ne->onReceive(c, c->GetData(), _reactorId);
            c->PopData(c->DataLen());
        }
    }


private:
    std::vector<Client*> _clientBuffer;
    std::map<int, Client*> _clients;
    bool _runFlag {false};
    std::mutex _mutex;
    std::condition_variable _cond;
    int _epollfd {0};
    int _mEvents {1024};
    NetEvent* _ne;
    int _exitStatus {0};
    epoll_event* _evs {nullptr};
    int _reactorId {0};
};

#endif